package com.avicit.zipkinkafkaesserver.service;

import com.avicit.zipkinkafkaesserver.config.ConfigDB;
import com.avicit.zipkinkafkaesserver.config.ConfigDBInterfaceCount;
import com.avicit.zipkinkafkaesserver.config.JobPropertiesConfig;
import com.avicit.zipkinkafkaesserver.dao.ZipkinInterfaceCountDao;
import com.avicit.zipkinkafkaesserver.dao.es.ZipkinInterfaceCountEsDao;
import com.avicit.zipkinkafkaesserver.dao.es.entity.ZipkinInterfaceCountES;
import com.avicit.zipkinkafkaesserver.dto.ZipkinInterfaceCountDto;
import com.avicit.zipkinkafkaesserver.entity.ZipkinInterfaceCount;
import com.avicit.zipkinkafkaesserver.entity.ZipkinUpdateMsg;
import com.avicit.zipkinkafkaesserver.enums.ESIndexsEnum;
import com.avicit.zipkinkafkaesserver.enums.JobTypeEnum;
import com.avicit.zipkinkafkaesserver.factory.IDFactory;
import com.avicit.zipkinkafkaesserver.utils.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import scala.Tuple2;

import javax.persistence.criteria.Order;
import javax.persistence.criteria.Predicate;
import java.io.IOException;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 接口调用次数统计
 */
@Slf4j
@Service
public class ZipkinInterfaceCountService {

    @Autowired
    private ZipkinInterfaceCountEsDao zipkinInterfaceCountEsDao;

    @Autowired
    private ZipkinInterfaceCountDao zipkinInterfaceCountDao;
    @Autowired
    private ZipkinUpdateMsgService zipkinUpdateMsgService;

    @Autowired
    private JobPropertiesConfig jobPropertiesConfig;

//    /**
//     * 批量保存 到 es
//     * @param index
//     * @param type
//     * @param zipkinInterfaceCountList
//     * @return
//     */
//    public boolean save(String index, String type, List<> zipkinInterfaceCountList) {
//        return zipkinInterfaceCountEsDao.save(index, type, zipkinInterfaceCountList);
//    }

    public boolean save(String index, String type, List<ZipkinInterfaceCountES> zipkinInterfaceCountList) throws IOException {
        return zipkinInterfaceCountEsDao.save(index, type, zipkinInterfaceCountList);
    }

    /**
     * 将 spark 计算的结果保存到es
     * @param tuple2List
     * @return
     */
    public boolean save(List<Tuple2<String, Integer>>tuple2List) {
        try {
            List<ZipkinInterfaceCount> countVos = tuple2List.stream()
                    .map(tuple2 -> {
                        String[] split = tuple2._1.split("@");
                        return ZipkinInterfaceCount.builder()
//                                .id(IDFactory.uuid())
                                .interfaceIp(split[0])//取ip
                                .serverName(split[1])//取服务名
                                .interfacePath(split[3])//取接口路径
                                .interfaceStatus(split[4])//取接口状态
                                .interfaceCount(tuple2._2)//取统计结果
//                                .createTime(Long.valueOf(split[5]))//取创建时间
//                                .queryKey(new StringBuffer(split[0]).append("@")
//                                        .append(split[1]).append("@")
//                                        .append(split[2]).append("@")
//                                        .append(split[3]).append("@")
//                                        .append(split[4]).toString())
                                .build();
                    })
                    .collect(Collectors.toList());

            //批量写入es
            String indexPrefix = ESIndexsEnum.ZipkinInterfaceCountPrefix.getIndexPrefix();
            String index = new StringBuffer(indexPrefix).append("-").append(TimeUtils.nowDateString()).toString();
//            boolean save = this.save(index, indexPrefix, countVos);
//            if (!save) log.error("save ZipkinInterfaceCountList to es error :{}",false);
//            return save;
        }catch (Exception e) {
            log.error("save ZipkinInterfaceCountList to es error :{}",e.getLocalizedMessage());
            return false;
        }
        return false;
    }


    public List<ZipkinInterfaceCount> searchFromEs() {
//       return interfaceCounts();
        return null;
    }

    public Map<String, Double> interfaceCounts(long startTime, long endTime) {
        String indexPrefix = ESIndexsEnum.ZipkinInterfaceCountPrefix.getIndexPrefix();
        String index = new StringBuffer(indexPrefix).append("-").append(TimeUtils.nowDateString()).toString();
        Map<String, Double> interfaceCount = zipkinInterfaceCountEsDao.doInterfaceCount(startTime, endTime, index, indexPrefix);
        return interfaceCount;
    }

    /**
     * 保存结果到es
     * @param interfaceCount
     */
    public boolean saveToEs(List<Tuple2<String, Integer>> interfaceCount,Long startTime, Long endTime, Long jobStartTime, Long jobEndTime) {
        try {
//            //获取批次号
//            BatchNumberFactory batchNumberFactory = BatchNumberContext.batchNumberContext().batchNumberFactory(BatchNumberTypeEnum.ZipkinInterfaceCount);
//            Integer batchNumber = batchNumberFactory.batchNumber();

            List<ZipkinInterfaceCountES> countVos = interfaceCount.stream()
                    .map(tuple2 -> {
                        String[] split = tuple2._1.split("@");
                        return ZipkinInterfaceCountES.builder().id(IDFactory.uuid()).interfaceIp(split[0])//取ip
                                .serverName(split[1])//取服务名
                                .interfacePath(split[3])//取接口路径
                                .interfaceStatus(split[4])//取接口状态
                                .interfaceCount(tuple2._2)//取统计结果
//                                .batchNumber(batchNumber)//设置批次号
                                .esEndTime(endTime)//整个 结果集都是在同一个时间段的数据
                                .esStartTime(startTime)
                                .queryStartTime(jobStartTime)
                                .queryEndTime(jobEndTime)//记下记录产生的时间段
//                                .createTime(Long.valueOf(split[5]))//取创建时间
                                .queryKey(new StringBuffer(split[0])
                                        .append("@").append(split[1]).append("@").append(split[2]).append("@")
                                        .append(split[3]).append("@").append(split[4]).toString()).build();
                    }).collect(Collectors.toList());

            //批量写入es
            String type = type();
            String index = index();
            boolean save = this.save(index, type, countVos);
            if (!save) log.error("save ZipkinInterfaceCountList to es error :{}",false);
            return save;
        }catch (Exception e) {
            log.error("save ZipkinInterfaceCountList to es error :{}",e.getLocalizedMessage());
            return false;
        }
    }


    /**
     * 获取开始时间
     * 如果 根据最大的批次号能获取到数据，就取最新的结束时间作为下一次的开始时间
     * 否则就是第一次运行，直接取配置文件的开始时间
     * @return
     */
    public Long initStartTime() throws IOException {
        ZipkinInterfaceCountES zipkinInterfaceCountES = zipkinInterfaceCountEsDao.findByBatchNumber(index(),type());
        if (zipkinInterfaceCountES.getEsEndTime() == null) {//当天的没有，查询昨天的
            String yestarday = TimeUtils.backFor(1);
            zipkinInterfaceCountES = zipkinInterfaceCountEsDao.findByBatchNumber(index(yestarday),type());
        }
        ConfigDBInterfaceCount configDB = jobPropertiesConfig.getConfigDBInterfaceCount();

        //是否是系统启动状态
        String startTimeString = configDB.isStatus()?configDB.getStartTime(): new StringBuffer(TimeUtils.nowDateString()).append(" ").append("00:00:00").toString();
        long startTimeFromConfig = TimeUtils.dateTimeToMilliSecond(startTimeString, TimeUtils.DatePattern.ALLTIME);
        long currentTimeMillis = System.currentTimeMillis();
        long startTime = startTimeFromConfig>= currentTimeMillis ? currentTimeMillis :startTimeFromConfig;
        return zipkinInterfaceCountES.getEsEndTime() == null//这里有两种情况是null ，1 是系统刚启动，2是时间到新的一天。如果是新的一天，configDB.isStatus() = false ，开始时间取00:00:00
                ? startTime
                : zipkinInterfaceCountES.getEsEndTime()+1;//防止边界重复
    }


    public void updateStartTime(long endTime) throws IOException {
        ZipkinInterfaceCountES zipkinInterfaceCountES = zipkinInterfaceCountEsDao.findByBatchNumber(index(),type());
        if (zipkinInterfaceCountES.getId() == null) {//没有查询，就新增
            log.info("update zipkinInterfaceCountES id is null and save endTime :{}",TimeUtils.timestamToDatetime(endTime));
            log.info("current endTime :{}",endTime);
            zipkinInterfaceCountES = ZipkinInterfaceCountES.builder().build();
            zipkinInterfaceCountES.setEsEndTime(endTime);
            zipkinInterfaceCountEsDao.save(index(),type(),zipkinInterfaceCountES);
            return;
        }
        if (endTime>=System.currentTimeMillis()) {log.error("current endTime :{}",TimeUtils.timestamToDatetime(endTime));return;}
        Map<String, Object> updateMap = new HashMap<>();
        updateMap.put("esEndTime", endTime);
        log.info("update zipkinInterfaceCountES  endTime :{}",TimeUtils.timestamToDatetime(endTime));
        zipkinInterfaceCountEsDao.update(updateMap,index(),type(),zipkinInterfaceCountES.getId());
    }





    /**
     * 从es 获取数据，统计结果，并返回
     * @return
     */
    public List<ZipkinInterfaceCount> getCountResultFromEs() {
        //es 聚合查询，统计一天的调用总次数，返回，今天统计昨天的
        String indexPrefix = ESIndexsEnum.ZipkinInterfaceCountPrefix.getIndexPrefix();
        LocalDate yesterday = LocalDate.now().plusDays(-1);//昨天的日期
        String index = new StringBuffer(indexPrefix).append("-").append(yesterday).toString();
        List<ZipkinInterfaceCountDto> countResultFromEs = zipkinInterfaceCountEsDao.getCountResultFromEs(index, type());
        List<ZipkinInterfaceCount> collect = countResultFromEs.stream().filter(item->!StringUtils.isEmpty(item.getInterfaceKey())).map(zipkinInterfaceCountDto -> {
            String interfaceKey = zipkinInterfaceCountDto.getInterfaceKey();
            //ok /error 次数统计
            Integer interfaceCount = zipkinInterfaceCountDto.getInterfaceCount();//总的调用次数
            String[] split = interfaceKey.split("@");
            ZipkinInterfaceCount count = ZipkinInterfaceCount.builder()
                    .interfaceCount(interfaceCount)
                    .serverName(split[0])//服务名
                    .interfaceMethods(split[1])//请求方式
                    .interfacePath(split[2])//请求路径
                    .queryDate(yesterday.toString())//设置该批次的所属日期 ：当前日期的前一天
                    .build();
            Map<String, Integer> errorOk = zipkinInterfaceCountDto.getErrorOk();
            //设置正确的调用次数和错误的调用次数
            errorOk.entrySet().forEach(kv->{
                String key = kv.getKey();
                Integer value = kv.getValue();
                if ("error".equals(key)) {
                    count.setCountError(value);
                }
                if ("ok".equals(key)) {
                    count.setOk(value);
                }
            });
            return count;
        }).collect(Collectors.toList());
        return collect;
    }

    /**
     * 将数据写入数据库
     * @param countResultFromEs
     */
    public void saveAllToDb(List<ZipkinInterfaceCount> countResultFromEs) {
        try {
            List<ZipkinInterfaceCount> result = new ArrayList<>();
            countResultFromEs.forEach(item->{
                ZipkinInterfaceCount zipkinInterfaceCount = zipkinInterfaceCountDao.findByItems(item.getServerName(),item.getQueryDate(),item.getInterfacePath(),item.getInterfaceMethods());
                if (StringUtils.isEmpty(zipkinInterfaceCount)) {
                    item.setId(IDFactory.uuid());
                    result.add(item);
                }else {
                    zipkinInterfaceCount.setOk(item.getOk());
                    zipkinInterfaceCount.setCountError(item.getCountError());
                    zipkinInterfaceCountDao.save(zipkinInterfaceCount);
                }
            });
            zipkinInterfaceCountDao.saveAll(result);
           // this.doSaveUpdateMsg(countResultFromEs);
        }catch (Exception e) {
            log.error("save interface count msg error : {}", e.getLocalizedMessage());
            //doErrorSaveUpdateMsg(countResultFromEs);
        }

    }

    /**
     * 接口统计信息批量成功写入时，记录 更新信息
     * @param countResultFromEs
     */
    private void doSaveUpdateMsg(List<ZipkinInterfaceCount> countResultFromEs) {
        if (countResultFromEs.size() == 0) {return;}
        String queryDate = countResultFromEs.get(0).getQueryDate();
        ZipkinUpdateMsg build = ZipkinUpdateMsg.builder().build();
        build.setJobType("zipkin_interface_count");
        build.setUpdateStatus(true);//该批次的接口统计次数写入成功
        build.setQueryDate(queryDate);
        build.setEsIndex(index(queryDate));
        build.setEsType(type());
        zipkinUpdateMsgService.save(build);
    }

    /**
     * 更新异常记录 信息
     * @param countResultFromEs
     */
    private void doErrorSaveUpdateMsg(List<ZipkinInterfaceCount> countResultFromEs) {
        if (countResultFromEs.size() == 0) {return;}
        String jobType = JobTypeEnum.ZipkinInterfaceCount.getDescription();
        //属于哪天的数据
        String queryDate = countResultFromEs.get(0).getQueryDate();
        ZipkinUpdateMsg zipkinUpdateMsg = zipkinUpdateMsgService.findByQueryDateAndUpdateStatus(queryDate, jobType,false);
        if (StringUtils.isEmpty(zipkinUpdateMsg)) {//当前日期没有异常的更新
            ZipkinUpdateMsg build = ZipkinUpdateMsg.builder().build();
            build.setJobType("zipkin_interface_count");
            build.setUpdateStatus(false);//该批次的接口统计次数写入失败
            build.setQueryDate(queryDate);
            build.setEsIndex(index(queryDate));
            build.setErrorCount(1);
            build.setEsType(type());
            zipkinUpdateMsgService.save(build);
            return;
        }
        //错误次数加一
        zipkinUpdateMsg.setErrorCount(zipkinUpdateMsg.getErrorCount()+1);
        zipkinUpdateMsgService.save(zipkinUpdateMsg);
    }


    /**
     * index 按天 创建
     * @param date
     * @return
     */
    private String index(String date) {
        final String index = new StringBuffer(type()).append("-").append(date).toString();
        return index;
    }

    /**
     * index 的前缀作为 type
     * @return
     */
    private String type() {
        final String indexPrefix = ESIndexsEnum.ZipkinInterfaceCountPrefix.getIndexPrefix();
        return indexPrefix;
    }

    /**
     * index 按天 创建,默认当前日期
     * @return
     */
    private String index() {
        final String index = new StringBuffer(type()).append("-").append(TimeUtils.nowDateString()).toString();
        return index;
    }


    //页面接口,更具日期查询
    public List<ZipkinInterfaceCount> interfaceCountsv2(String queryDate) {
        return zipkinInterfaceCountDao.findAll((Specification<ZipkinInterfaceCount>) (root, criteriaQuery, criteriaBuilder) -> {
            Predicate predicateRoot = null;
            if (!StringUtils.isEmpty(queryDate)) {
                predicateRoot = criteriaBuilder.equal(root.get("queryDate"), queryDate);
            }
            return predicateRoot;
        });
    }

    public List<ZipkinInterfaceCount> interfaceCountsv2(final String queryDateFrom, final String queryDateTo, final String serverName, final String interfacePath) {
        return zipkinInterfaceCountDao.findAll((Specification<ZipkinInterfaceCount>) (root, criteriaQuery, criteriaBuilder) -> {

            Predicate predicateRoot = criteriaBuilder.greaterThanOrEqualTo(root.get("queryDate"),queryDateFrom);//大于等于
            Predicate lessThanOrEqualTo = criteriaBuilder.lessThanOrEqualTo(root.get("queryDate"), queryDateTo);//小于等于
            predicateRoot = criteriaBuilder.and(predicateRoot,lessThanOrEqualTo);//取区间
            if (!StringUtils.isEmpty(serverName)) {//服务名
                Predicate serverNamePredicate = criteriaBuilder.equal(root.get("serverName").as(String.class), serverName);
                predicateRoot = criteriaBuilder.and(predicateRoot,serverNamePredicate);
            }
            if (!StringUtils.isEmpty(interfacePath)) {//接口
                predicateRoot = criteriaBuilder.and(predicateRoot,criteriaBuilder.equal(root.get("interfacePath").as(String.class),interfacePath));
            }
            //接口调用次数倒排
            Order desc = criteriaBuilder.desc(root.get("interfaceCount"));
            criteriaQuery.where(predicateRoot).orderBy(desc);
            return null;
        }, PageRequest.of(0,100)).getContent();
    }
}
